package net.maergaiyun.func;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import net.maergaiyun.model.ShortLinkWideDO;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.http.HttpEntity;
import org.apache.http.HttpStatus;
import org.apache.http.client.HttpClient;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.config.Registry;
import org.apache.http.config.RegistryBuilder;
import org.apache.http.conn.socket.ConnectionSocketFactory;
import org.apache.http.conn.socket.PlainConnectionSocketFactory;
import org.apache.http.conn.ssl.SSLConnectionSocketFactory;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClientBuilder;
import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
import org.apache.http.util.EntityUtils;

@Slf4j
public class LocationMapFunction extends RichMapFunction<ShortLinkWideDO, String> {
    private CloseableHttpClient httpClient;

    private static final String IP_PARSE_URL = "https://restapi.amap.com/v3/ip?ip=%s&output=json&key=995b62fbaef37f1eee3ad94f54a36f8d";

    @Override
    public String map(ShortLinkWideDO shortLinkWideDO) throws Exception {
        String ip = shortLinkWideDO.getIp();
        String url = String.format(IP_PARSE_URL, ip);
        HttpGet httpGet = new HttpGet(url);

        try (CloseableHttpResponse response = httpClient.execute(httpGet)) {
            int statusCode = response.getStatusLine().getStatusCode();
            if (statusCode == HttpStatus.SC_OK) {
                HttpEntity entity = response.getEntity();
                String result = EntityUtils.toString(entity, "UTF-8");
                JSONObject locationObj = JSON.parseObject(result);

                String province = locationObj.getString("province");
                String city = locationObj.getString("city");
                shortLinkWideDO.setProvince(province);
                shortLinkWideDO.setCity(city);
            }
        } catch (Exception e) {
            log.error("ip解析错误, shortLinkWideDO={}, e={}", shortLinkWideDO, e.getMessage());
        }

        return JSONObject.toJSONString(shortLinkWideDO);
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        this.httpClient = createHttpClient();
    }

    @Override
    public void close() throws Exception {
        if (this.httpClient != null) {
            this.httpClient.close();
        }
    }

    public CloseableHttpClient createHttpClient() {
        Registry<ConnectionSocketFactory> registry = RegistryBuilder.<ConnectionSocketFactory>create().register("http", PlainConnectionSocketFactory.getSocketFactory()).register("https", SSLConnectionSocketFactory.getSocketFactory()).build();
        PoolingHttpClientConnectionManager connectionManager = new PoolingHttpClientConnectionManager(registry);

        // 设置连接池最大500个连接
        connectionManager.setMaxTotal(500);
        // MaxPerRoute是对MaxTotal的细分, 每个主机的最大并发是300, route是指域名
        connectionManager.setDefaultMaxPerRoute(300);
        /**
         *
         * 只请求maergeiyun.net最大并发是300
         *
         * 请求maergeiyun.net最大并发是300
         * 请求openai.com最大并发是200
         */

        RequestConfig requestConfig = RequestConfig.custom()
                // 返回数据的超时时间
                .setSocketTimeout(20000)
                // 连接上服务器的超时时间
                .setConnectTimeout(10000)
                // 从连接池中获取连接的超时时间
                .setConnectionRequestTimeout(1000).build();

        CloseableHttpClient closeableHttpClient = HttpClientBuilder.create().setDefaultRequestConfig(requestConfig).setConnectionManager(connectionManager).build();

        return closeableHttpClient;
    }
}
